package cn.itcast.storm;

import java.util.Map;
import java.util.Random;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

public class RandomSpout extends BaseRichSpout {
	
	SpoutOutputCollector collector = null;
	String[] goods = {"iphone","xiaomi","meizu","zhongxing","huawei","moto","sumsung","simens"};

	/**
	 * 获取消息并发送下一个组件的方法，会被storm不断地调用
	 * 从goods数据中随机取一个商品名称封装到tuple中发送出去
	 */
	@Override
	public void nextTuple() {
		
		//随机取到一个商品名称
		Random random = new Random();
		String good = goods[random.nextInt(goods.length)];
		
		//封装到tuple中发送出去
		collector.emit(new Values(good));
		//休眠500毫秒
		Utils.sleep(500);
		
	}

	/**
	 * 进行初始化，只在开始的时候调用一次
	 */
	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		
		this.collector = collector;
		
	}

	/**
	 * 定义tuple的scheme
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("src_word"));
		
	}

}
